package com.atguigu.flink.processFunction;

import com.atguigu.flink.bean.WaterSensor;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class TopNWindowkeyBy {
    public static void main(String[] args) throws Exception {
        //TODO 1.准备环境
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //TODO 2.从指定的网络端口中读取数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.对流中数据进行类型转换    String->WaterSensor
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS
                .map(new WaterSensorMapFunction())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner(
                                        new SerializableTimestampAssigner<WaterSensor>() {
                                            @Override
                                            public long extractTimestamp(WaterSensor ws, long recordTimestamp) {
                                                return ws.getTs() * 1000;
                                            }
                                        }
                                )
                );

        //TODO 4.分组
        KeyedStream<WaterSensor, Integer> keyedDS = wsDS.keyBy(WaterSensor::getVc);
        
        //TODO 5.开窗
        WindowedStream<WaterSensor, Integer, TimeWindow> windowDS = keyedDS.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //TODO 6.对窗口数据进行处理
        SingleOutputStreamOperator<Tuple3<Integer, Integer, Long>> aggregateDS = windowDS.aggregate(
                new AggregateFunction<WaterSensor, Integer, Integer>() {
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    @Override
                    public Integer add(WaterSensor waterSensor, Integer accumulator) {
                        return ++accumulator;
                    }

                    @Override
                    public Integer getResult(Integer accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Integer merge(Integer integer, Integer acc1) {
                        return 0;
                    }
                },
                new WindowFunction<Integer, Tuple3<Integer, Integer, Long>, Integer, TimeWindow>() {
                    @Override
                    public void apply(Integer vc, TimeWindow window, Iterable<Integer> input, Collector<Tuple3<Integer, Integer, Long>> out) throws Exception {
                        Integer count = input.iterator().next();
                        long endTime = window.getEnd();
                        out.collect(Tuple3.of(vc, count, endTime));
                    }

                }

        );
        //TODO 7.分组
        KeyedStream<Tuple3<Integer, Integer, Long>, Long> endKeyDS = aggregateDS.keyBy(t -> t.f2);

        //TODO 8.对分组数据进行处理
        endKeyDS.process(
                new KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String >() {
                    Map<Long , List<Tuple3<Integer,Integer,Long>>> vcCountMap = new HashMap<>();
                    @Override
                    public void processElement(Tuple3<Integer, Integer, Long> value, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.Context ctx, Collector<String> out) throws Exception {
                        Long end = value.f2;
                        if (vcCountMap.containsKey(end)){
                            vcCountMap.get(end).add(value);
                        } else {
                            List<Tuple3<Integer, Integer, Long>> vcCountList = new ArrayList<>();
                            vcCountList.add(value);
                            vcCountMap.put(end , vcCountList);
                        }
                        TimerService timerService = ctx.timerService();
                        timerService.registerEventTimeTimer(end + 1);
                    }

                    @Override
                    public void onTimer(long timestamp, KeyedProcessFunction<Long, Tuple3<Integer, Integer, Long>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        Long end = ctx.getCurrentKey();
                        List<Tuple3<Integer, Integer, Long>> vcCountList = vcCountMap.get(end);
                        vcCountList.sort((o1 , o2) -> o1.f1 - o2.f1);

                        StringBuilder outStr = new StringBuilder();

                        outStr.append("================================\n");
                        // 遍历 排序后的 List，取出前 threshold 个， 考虑可能List不够2个的情况  ==》 List中元素的个数 和 2 取最小值
                        for (int i = 0; i < Math.min(2, vcCountList.size()); i++) {
                            Tuple3<Integer, Integer, Long> vcCount = vcCountList.get(i);
                            outStr.append("Top" + (i + 1) + "\n");
                            outStr.append("vc=" + vcCount.f0 + "\n");
                            outStr.append("count=" + vcCount.f1 + "\n");
                            outStr.append("窗口结束时间=" + vcCount.f2 + "\n");
                            outStr.append("================================\n");
                        }

                        // 用完的List，及时清理，节省资源
                        vcCountList.clear();

                        out.collect(outStr.toString());
                    }
                }
        );
        //TODO 6.打印输出
        //processDS.print();
        //TODO 7.提交作业
        env.execute();

    }
}
